package cn.edu.flink.scala.tutorial.cep

import org.apache.flink.cep.PatternSelectFunction
import org.apache.flink.cep.scala.CEP
import org.apache.flink.cep.scala.pattern.Pattern
import org.apache.flink.streaming.api.scala._

import java.util

object CEPTest {
  case class LoginLog(userid: String, ip: String, eventType: String, timestamp: Long)

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI()
    env.setParallelism(1)

    val loginStream = env.fromElements(
      LoginLog("user_1", "192.168.0.1", "fail", 2000L),
      LoginLog("user_1", "192.168.0.1", "fail", 3000L),
      LoginLog("user_2", "192.168.0.1", "fail", 4000L),
      LoginLog("user_1", "192.168.0.1", "fail", 5000L),
      LoginLog("user_2", "192.168.0.1", "fail", 6000L),
      LoginLog("user_2", "192.168.0.1", "success", 7000L),
      LoginLog("user_1", "192.168.0.1", "fail", 8000L)
    ).assignAscendingTimestamps(_.timestamp)
      .keyBy(_.userid)

    // 定义pattern, 连续三次登陆失败事件
    val pattern = Pattern
      .begin[LoginLog]("first")
      .where(_.eventType.equals("fail"))
      .next("second")
      .where(_.eventType.equals("fail"))
      .next("third")
      .where(_.eventType.equals("fail"))

    val patternStream = CEP.pattern(loginStream, pattern)

    patternStream
      .select(new PatternSelectFunction[LoginLog, String] {
        override def select(map: util.Map[String, util.List[LoginLog]]): String = {
          val first = map.get("first").get(0)
          val second = map.get("second").get(0)
          val third = map.get("third").get(0)
          s"${first.userid} 连续三次登录失败！ 登陆时间： ${first.timestamp}, ${second.timestamp}, ${third.timestamp}"
        }
      })
      .print("warning")

    env.execute("CEPTest")
  }
}
